package com.nepu.springboot.springbootrealtime.mapper.imp;

import com.nepu.springboot.springbootrealtime.bean.NameValue;
import com.nepu.springboot.springbootrealtime.mapper.PublisherMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.lucene.search.TotalHits;
import org.elasticsearch.ElasticsearchStatusException;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.text.Text;
import org.elasticsearch.index.query.MatchQueryBuilder;
import org.elasticsearch.index.query.Operator;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.aggregations.Aggregation;
import org.elasticsearch.search.aggregations.AggregationBuilder;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.Aggregations;
import org.elasticsearch.search.aggregations.bucket.terms.ParsedStringTerms;
import org.elasticsearch.search.aggregations.bucket.terms.ParsedTerms;
import org.elasticsearch.search.aggregations.bucket.terms.Terms;
import org.elasticsearch.search.aggregations.bucket.terms.TermsAggregationBuilder;
import org.elasticsearch.search.aggregations.metrics.ParsedSum;
import org.elasticsearch.search.aggregations.metrics.SumAggregationBuilder;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.fetch.subphase.highlight.HighlightBuilder;
import org.elasticsearch.search.fetch.subphase.highlight.HighlightField;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Repository;

import java.io.IOException;
import java.time.LocalDate;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @author chenshuaijun
 * @create 2022-11-06 10:44
 */
@Slf4j
@Repository
public class PublisherMapperImp implements PublisherMapper {
    @Autowired
    private RestHighLevelClient esClient;

    private String dauRealTimePrefix = "gmall_dau_info_";
    private String orderRealTimePrefix = "gmall_order_wide_";

    /**
     * 处理日活和每小时日活人数的对比
     * @param td 统计时间
     * @return
     */
    @Override
    public Map<String, Object> dauRealTime(String td) {
        Map<String, Object> dauMap = new HashMap<>();
        // 查询截止到此刻的日活
        Long dauTotal = searchDauRealTime(td);
        dauMap.put("dauTotal", dauTotal);
        // 查询今日分时的日活
        Map<String, Long> dauTd = timeSharingStatistics(td);
        dauMap.put("dauTd",dauTd);
        // 查询昨日的日活
        // 日期处理
        LocalDate yesterdayDate = LocalDate.parse(td).minusDays(1);
        Map<String, Long> dauYd = timeSharingStatistics(yesterdayDate.toString());
        dauMap.put("dauYd",dauYd);
        return dauMap;
    }

    /**
     * 统计图
     * @param itemName 商品名称
     * @param date 日期
     * @param fieldName 聚合属性
     * @return 值
     */
    @Override
    public List<NameValue> statsByItem(String itemName, String date, String fieldName) {
        List<NameValue> result = searchByItem(itemName, date, fieldName);
        return result;
    }

    @Override
    public Map<String, Object> detailByItem(String itemName, String date, String pageNo, String pageSize) {
        Map<String, Object> result = esDetailByItem(itemName, date, pageNo, pageSize);
        return result;
    }

    /**
     * 统计当日的日活
     * @param td 统计日期
     * @return 截止到当前的日活的人数
     */
    private Long searchDauRealTime(String td){

        // 构建查询的key
        String indexName = dauRealTimePrefix+td;
        try {
            SearchRequest searchRequest = new SearchRequest(indexName);
            SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
            // 扫描之后，直接丢弃掉明细数据
            searchSourceBuilder.size(0);
            searchRequest.source(searchSourceBuilder);
            SearchResponse searchResponse = esClient.search(searchRequest, RequestOptions.DEFAULT);
            // 处理结果
            return searchResponse.getHits().getTotalHits().value;
        } catch (ElasticsearchStatusException es){
            log.warn(indexName + "-->索引不存在");
        } catch (IOException e) {
            e.printStackTrace();
            throw new RuntimeException("运行时异常.....");
        }
        // 如果发生了异常，直接现实没数据
        return 0L;
    }

    /**
     * 分时进行日活的统计
     * @param td 统计的日期
     * @return 分时统计结果
     */
    private Map<String, Long> timeSharingStatistics(String td){
        // 拼接查询es的index
        String indexName = dauRealTimePrefix+td;
        // 缓存结果
        Map<String, Long> resultMap = new HashMap<>();
        // 查询
        try {
            SearchRequest searchRequest = new SearchRequest(indexName);
            SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
            TermsAggregationBuilder termsAggregationBuilder = AggregationBuilders.terms("grouphour").field("hr").size(24);
            searchSourceBuilder.aggregation(termsAggregationBuilder);
            searchRequest.source(searchSourceBuilder);
            SearchResponse searchResponse = esClient.search(searchRequest, RequestOptions.DEFAULT);
            // 处理结果
            ParsedStringTerms aggregations = searchResponse.getAggregations().get("grouphour");
            List<? extends Terms.Bucket> buckets = aggregations.getBuckets();
            for (Terms.Bucket bucket : buckets) {
                String key = bucket.getKeyAsString();
                long docCount = bucket.getDocCount();
                resultMap.put(key,docCount);
            }
            return resultMap;
        } catch (ElasticsearchStatusException es){
            log.warn(indexName + "-->索引不存在");
        } catch (IOException e) {
            e.printStackTrace();
            throw new RuntimeException("运行时异常.....");
        }
        return resultMap;
    }

    private List<NameValue> searchByItem(String itemName, String date, String fieldName){
        // 处理key
        String indexName = orderRealTimePrefix+date;
        // 缓存结果
        List<NameValue> nameValues = new ArrayList<>();
        try {
            SearchRequest searchRequest = new SearchRequest(indexName);
            SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
            // 不要明细
            searchSourceBuilder.size(0);
            MatchQueryBuilder queryBuilder = QueryBuilders.matchQuery("sku_name", itemName).operator(Operator.AND);
            searchSourceBuilder.query(queryBuilder);
            TermsAggregationBuilder termsAggregationBuilder = AggregationBuilders.terms("fieldgroup").field(fieldName).size(100);
            SumAggregationBuilder sumAggregationBuilder = AggregationBuilders.sum("fieldsum").field("split_total_amount");
            termsAggregationBuilder.subAggregation(sumAggregationBuilder);
            searchSourceBuilder.aggregation(termsAggregationBuilder);
            searchRequest.source(searchSourceBuilder);
            SearchResponse searchResponse = esClient.search(searchRequest, RequestOptions.DEFAULT);
            // 处理结果
            ParsedTerms aggregations = searchResponse.getAggregations().get("fieldgroup");
            List<? extends Terms.Bucket> buckets = aggregations.getBuckets();
            for (Terms.Bucket bucket : buckets) {
                String key = bucket.getKey().toString();
                ParsedSum fieldsum = bucket.getAggregations().get("fieldsum");
                double value = fieldsum.getValue();
                NameValue nameValue = new NameValue(key, value);
                nameValues.add(nameValue);
            }
            return nameValues;
        } catch (ElasticsearchStatusException es){
            log.warn(indexName + "-->索引不存在");
        } catch (IOException e) {
            e.printStackTrace();
            throw new RuntimeException("运行时异常.....");
        }
        return nameValues;
    }

    private Map<String, Object> esDetailByItem(String itemName, String date, String pageNo, String pageSize){
        // 封装查询的索引
        String indexName = orderRealTimePrefix+date;
        HashMap<String, Object> resultMap = new HashMap<>();
        List<Map<String, Object>> resultList = new ArrayList<>();
        try {
            SearchRequest searchRequest = new SearchRequest(indexName);
            SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
            MatchQueryBuilder queryBuilder = QueryBuilders.matchQuery("sku_name", itemName).operator(Operator.AND);
            searchSourceBuilder.from(Integer.parseInt(pageNo)).size(Integer.parseInt(pageSize));
            HighlightBuilder highlightBuilder = new HighlightBuilder();
            highlightBuilder.field("sku_name");
            searchSourceBuilder.highlighter(highlightBuilder);
            searchSourceBuilder.fetchSource(new String[]{"create_time","order_price","province_name","sku_name","sku_num","total_amount","user_age","user_gender"},null);
            searchSourceBuilder.query(queryBuilder);
            searchRequest.source(searchSourceBuilder);
            SearchResponse searchResponse = esClient.search(searchRequest, RequestOptions.DEFAULT);
            // 处理结果
            // 获取总数据条数
            long total = searchResponse.getHits().getTotalHits().value;
            resultMap.put("total", total);
            SearchHit[] hits = searchResponse.getHits().getHits();
            for (SearchHit hit : hits) {
                Map<String, Object> sourceAsMap = hit.getSourceAsMap();
                // 提取高亮
                HighlightField highlightField = hit.getHighlightFields().get("sku_name");
                Text[] fragments = highlightField.getFragments();
                String highTag = fragments[0].toString();
                // 替换数据
                sourceAsMap.put("sku_name",highTag);
                resultList.add(sourceAsMap);
            }
            resultMap.put("detail", resultList);
        } catch (ElasticsearchStatusException es){
            log.warn(indexName + "-->索引不存在");
        } catch (IOException e) {
            e.printStackTrace();
            throw new RuntimeException("运行时异常.....");
        }

        return resultMap;
    }
}
